package com.cloud.point.concurrent.general;

import java.util.LinkedList;
import java.util.UUID;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * 异步消费模式--消息队列
 * -- 基于 wait/notify 实现
 *
 * @author 24373
 * @version 1.0
 * @date 2024-07-22 23:00
 */
@Slf4j
public class TestMessageQueue {

    public static void main(String[] args) throws InterruptedException {
        MessageQueue queue = new MessageQueue(10);
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                for (; ; ) {
                    String id = UUID.randomUUID().toString();
                    Message msg = new Message(id, "content" + id);
                    queue.put(msg);
                }
            }, "producer" + i).start();
        }
        new Thread(() -> {
            for (; ; ) {
                queue.take(500);
            }
        }, "consumer").start();

    }

    @AllArgsConstructor
    @Data
    static class Message {
        private String id;
        private Object message;

    }

    static class MessageQueue {

        private final int capacity;
        private final LinkedList<Message> queue;

        public MessageQueue(int capacity) {
            this.capacity = capacity;
            queue = new LinkedList<>();
        }

        public void put(Message message) {
            synchronized (queue) {
                while (queue.size() == capacity) {
                    log.info("队列满了，等待消费。线程：{}", Thread.currentThread().getId());
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        log.info("{}", e);
                    }
                }
                queue.addLast(message);
                log.info("唤醒消费者；生产消息{}，线程{}", message, Thread.currentThread().getId());
                queue.notifyAll();
            }
        }

        public Message take(long timeout) {
            synchronized (queue) {
                long star = System.currentTimeMillis();
                long pass = 0;
                while (queue.isEmpty()) {
                    try {
                        long waitTime = timeout - pass;
                        //等待超时，直接返回
                        if (waitTime <= 0) {
                            return null;
                        }
                        log.info("等待获取消息，线程{}", Thread.currentThread().getId());
                        queue.wait(waitTime);
                    } catch (InterruptedException e) {
                        log.info("{}", e);
                    } finally {
                        pass = System.currentTimeMillis() - star;
                    }
                }
                Message message = queue.removeFirst();
                log.info("唤醒生产者；获取消息{}，线程{}", message, Thread.currentThread().getId());
                queue.notifyAll();
                return message;
            }
        }
    }

}
